﻿
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Logging;
using Mysoft.Common.Enum;
using Mysoft.Tasks.Queue.IService;
using Mysoft.Tasks.Queue.IService.Attributes;
using Mysoft.Tasks.Queue.IService.Events;
using Mysoft.Tasks.Queue.IService.IEvents;
using Mysoft.Tasks.Queue.RabbitMQ.Extensions;
using Mysoft.Tasks.Queue.RabbitMQ.Factories;
using Mysoft.Tasks.Queue.RabbitMQ.Modules;
using RabbitMQ.Client;
using RabbitMQ.Client.Events;
using System.Text;

namespace Mysoft.Tasks.Queue.RabbitMQ
{
    /// <summary>
    /// RabbitMQ事件总线
    /// </summary>
    internal class RabbitMQEventBus : IMQEventBus
    {
        private readonly IRabbitMQPersistentConnection _persistentConnection;
        private readonly ILogger<RabbitMQEventBus> _logger;
        private readonly IServiceProvider _serviceProvider;
        private readonly IEventHandlerModuleFactory _eventHandlerFactory;
        /// <summary>
        /// 
        /// </summary>
        /// <param name="persistentConnection"></param>
        /// <param name="serviceProvider"></param>
        /// <param name="eventHandlerFactory"></param>
        /// <param name="logger"></param>
        public RabbitMQEventBus(IRabbitMQPersistentConnection persistentConnection, IServiceProvider serviceProvider, IEventHandlerModuleFactory eventHandlerFactory, ILogger<RabbitMQEventBus> logger)
        {
            _persistentConnection = persistentConnection ?? throw new ArgumentNullException(nameof(persistentConnection));
            _serviceProvider = serviceProvider ?? throw new ArgumentNullException(nameof(serviceProvider));
            _eventHandlerFactory = eventHandlerFactory ?? throw new ArgumentNullException(nameof(eventHandlerFactory));
            _logger = logger ?? throw new ArgumentNullException(nameof(logger));
        }

        private IModel _publishChannel;
        /// <summary>
        /// 发消息
        /// </summary>
        /// <typeparam name="TMessage"></typeparam>
        /// <param name="message">消息体</param>
        /// <param name="exchange">交换机</param>
        /// <param name="routingKey">路由</param>
        /// <param name="type">消息类型</param>
        /// <returns></returns>
        public void Publish<TMessage>(TMessage message, string exchange, string routingKey, string type = ExchangeType.Topic)
        {
            string body = message.Serialize();
            if (_publishChannel?.IsOpen != true)
            {
                if (_persistentConnection.IsConnected)
                {
                    _persistentConnection.TryConnect();
                }
                _publishChannel = _persistentConnection.ExchangeDeclare(exchange, type: type);
                _publishChannel.BasicReturn += async (se, ex) => await Task.Delay((int)_persistentConnection.Configuration.ConsumerFailRetryInterval.TotalMilliseconds).ContinueWith(t => Publish(body, ex.Exchange, ex.RoutingKey));
            }
            IBasicProperties properties = _publishChannel.CreateBasicProperties();
            properties.DeliveryMode = 2; // persistent
            _publishChannel.BasicPublish(exchange: exchange,
                             routingKey: routingKey,
                             mandatory: true,
                             basicProperties: properties,
                             body: body.GetBytes());
            _logger.WriteLog(_persistentConnection.Configuration.Level, $"{DateTimeOffset.Now.ToString("yyyy-MM-dd HH:mm:ss")}\t{exchange}\t{routingKey}\t{body}");
            _eventHandlerFactory?.PubliushEvent(new EventBusArgs(_persistentConnection.Endpoint, exchange, "", routingKey, type, _persistentConnection.Configuration.ClientProvidedName, body, true));
        }


        /// <summary>
        /// 发延时消息
        /// </summary>
        /// <typeparam name="TMessage"></typeparam>
        /// <param name="message">消息体</param>
        /// <param name="query">延时队列</param>
        /// <param name="exchange">延时交换机</param>
        /// <param name="routingKey">延时路由</param>
        /// <param name="exchange">死信交换机</param>
        /// <param name="routingKey">死信路由</param>
        /// <param name="delay">延时时间（毫秒）</param>
        /// <param name="QueryType">队列消息延时类型</param>
        /// <param name="type">队列消息交换机模式</param>
        /// <returns></returns>
        public void DelayedPublish<TMessage>(TMessage message, string queue, string exchange, string routingKey, string deadExchange, string deadRoutingKey, int delay, DeadQueryType QueryType = DeadQueryType.message, string type = ExchangeType.Topic) {
            string body = message.Serialize();
            if (_publishChannel?.IsOpen != true)
            {
                if (_persistentConnection.IsConnected)
                {
                    _persistentConnection.TryConnect();
                }
                _publishChannel = _persistentConnection.ExchangeDeclare(exchange, type: type);

                var arguments = new Dictionary<string, object>();
                arguments.Add("x-dead-letter-exchange", deadExchange);  //死信交换机
                arguments.Add("x-dead-letter-routing-key", deadRoutingKey);//死信路由
                if (QueryType == DeadQueryType.query) 
                {
                    arguments.Add("x-message-ttl", delay);
                }


                _publishChannel.QueueDeclare(queue: queue,
                                            durable: true,
                                            exclusive: false,
                                            autoDelete: false,
                                            arguments: arguments);

                _publishChannel.QueueBind(queue, exchange, routingKey, null);
                _publishChannel.BasicQos(0, 1, false);

                _publishChannel.BasicReturn += async (se, ex) => await Task.Delay((int)_persistentConnection.Configuration.ConsumerFailRetryInterval.TotalMilliseconds).ContinueWith(t => Publish(body, ex.Exchange, ex.RoutingKey));
            }
            IBasicProperties properties = _publishChannel.CreateBasicProperties();
            properties.DeliveryMode = 2; // persistent

            if (QueryType == DeadQueryType.message)
            {
                 properties.Expiration = delay.ToString();
            }

            _publishChannel.BasicPublish(exchange: exchange,
                             routingKey: routingKey,
                             mandatory: true,
                             basicProperties: properties,
                             body: body.GetBytes());
            _logger.WriteLog(_persistentConnection.Configuration.Level, $"{DateTimeOffset.Now.ToString("yyyy-MM-dd HH:mm:ss")}\t{exchange}\t{routingKey}\t{body}");
            _eventHandlerFactory?.PubliushEvent(new EventBusArgs(_persistentConnection.Endpoint, exchange, "", routingKey, type, _persistentConnection.Configuration.ClientProvidedName, body, true));

        }
        /// <summary>
        /// 订阅消息
        /// </summary>
        /// <param name="eventType">消息体</param>
        /// <param name="type">消息类型</param>
        public void Subscribe(Type eventType, string type = ExchangeType.Topic)
        {
            var attributes = eventType.GetCustomAttributes(typeof(EventBusAttribute), true);
            var millisecondsDelay = (int?)_persistentConnection?.Configuration?.ConsumerFailRetryInterval.TotalMilliseconds ?? 1000;
            foreach (var attribute in attributes)
            {
                if (attribute is EventBusAttribute attr)
                {
                    string queue = attr.Queue ?? $"{ attr.Exchange }.{ eventType.Name }";
                    if (!_persistentConnection.IsConnected)
                    {
                        _persistentConnection.TryConnect();
                    }
                    IModel channel;
                    #region snippet
                    try
                    {
                        channel = _persistentConnection.ExchangeDeclare(exchange: attr.Exchange, type: type);
                        channel.QueueDeclarePassive(queue);
                    }
                    catch
                    {
                        channel = _persistentConnection.ExchangeDeclare(exchange: attr.Exchange, type: type);
                        channel.QueueDeclare(queue: queue,
                                             durable: true,
                                             exclusive: false,
                                             autoDelete: false,
                                             arguments: null);
                    }
                    #endregion
                    channel.QueueBind(queue, attr.Exchange, attr.RoutingKey, null);
                    channel.BasicQos(0, 1, false);
                    EventingBasicConsumer consumer = new EventingBasicConsumer(channel);
                    consumer.Received += async (model, ea) =>
                    {
                        string body = Encoding.UTF8.GetString(ea.Body.ToArray());
                        bool isAck = false;
                        try
                        {
                            await ProcessEvent(body, eventType, ea);
                            channel.BasicAck(ea.DeliveryTag, multiple: false);
                            isAck = true;
                        }
                        catch (Exception ex)
                        {
                            _logger.LogError(new EventId(ex.HResult), ex, ex.Message);
                        }
                        finally
                        {
                            _eventHandlerFactory?.SubscribeEvent(new EventBusArgs(_persistentConnection.Endpoint, ea.Exchange, queue, attr.RoutingKey, type, _persistentConnection.Configuration.ClientProvidedName, body, isAck));
                            _logger.WriteLog(_persistentConnection.Configuration.Level, $"{DateTimeOffset.Now.ToString("yyyy-MM-dd HH:mm:ss")}\t{isAck}\t{ea.Exchange}\t{ea.RoutingKey}\t{body}");
                            if (!isAck)
                            {
                                await Task.Delay(millisecondsDelay);
                                channel.BasicNack(ea.DeliveryTag, false, true);
                            }
                        }
                    };
                    channel.CallbackException += (sender, ex) =>
                    {
                        _logger.LogError(new EventId(ex.Exception.HResult), ex.Exception, ex.Exception.Message);
                    };
                    channel.BasicConsume(queue: queue, autoAck: false, consumer: consumer);
                }
            }
        }
   
        /// <summary>
        /// 
        /// </summary>
        /// <param name="body"></param>
        /// <param name="eventType"></param>
        /// <param name="eventHandleType"></param>
        /// <param name="args"></param>
        /// <returns></returns>
        private async Task ProcessEvent(string body, Type eventType, BasicDeliverEventArgs args)
        {
            using (var scope = _serviceProvider.CreateScope())
            {
                foreach (Type eventHandleType in typeof(IEventHandler<>).GetMakeGenericType(eventType))
                {
                    object eventHandler = scope.ServiceProvider.GetRequiredService(eventHandleType);
                    if (eventHandler == null)
                    {
                        throw new InvalidOperationException(eventHandleType.Name);
                    }
                    Type concreteType = typeof(IEventHandler<>).MakeGenericType(eventType);
                    await (Task)concreteType.GetMethod(nameof(IEventHandler<IEvent>.Handle)).Invoke(
                        eventHandler,
                        new object[] {
                         Activator.CreateInstance(typeof(EventHandlerArgs<>).MakeGenericType(eventType), new object[] { body, args.Redelivered, args.Exchange, args.RoutingKey })
                        });
                }
            }
        }
        public string GetMode()
        {
            return "RabbitMQ";
        }
    }
}
